package com.lion.hbase.api;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.ParameterizedType;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.filter.Filter;

import com.lion.hbase.exception.HbaseClientException;
import com.lion.hbase.util.GenericsUtils;
import com.lion.hbase.util.ReflectUtil;
import com.lion.hbase.util.TimeUtil;
/**
 * 基础DAO，其他DAO需继承此DAO
 * @author 李永曜
 * 2015年6月17日
 */
public class HbaseDAO<T extends HbaseTableModel> {
	private Class<T> clazz;
	private HTableInfo<T> hTableInfo;
	private static final Log LOG = LogFactory.getLog(HbaseDAO.class);

	@SuppressWarnings("unchecked")
	public HbaseDAO() {
		this.clazz = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
		try {
			hTableInfo = new HTableInfo<T>(clazz);
		} catch (HbaseClientException e) {
			LOG.error(e.getMessage(), e);
		}
	}
	/**
	 * 获得当前实例对应的表名
	 * @return
	 * @throws IOException
	 */
	public HTableInterface getTableName() throws IOException {
		return HbaseAPI.getHtable(hTableInfo.getTableName());
	}
	/**
	 * 获得hbase中列名,一列对应一个成员变量
	 * @param kv
	 * @return
	 */
	private String getfieldName(KeyValue kv){
		return KeyValue.keyToString(kv.getKey()).split(":")[1].split("/")[0];
	}
	/**
	 * 将kv赋值到对象成员变量
	 * @param kv
	 * @param t
	 * @throws IllegalArgumentException
	 * @throws IllegalAccessException
	 * @throws SecurityException
	 * @throws NoSuchFieldException
	 * @throws InvocationTargetException 
	 * @throws NoSuchMethodException 
	 */
	private void parseKeyValue(KeyValue kv,T t) throws IllegalArgumentException, IllegalAccessException, SecurityException, NoSuchFieldException, NoSuchMethodException, InvocationTargetException{
		if (kv.matchingFamily(hTableInfo.getFamilyName().getBytes())) {
			String fieldName = getfieldName(kv);
			String value = new String(kv.getValue(), Charset.forName("UTF-8"));
			HFieldInfo fieldInfo = hTableInfo.getBind().get(fieldName);
			if (fieldInfo != null) {
				Field field = fieldInfo.getField();
				if(field.getType() == String.class){
					ReflectUtil.setValueByNameInokeMethod(field, value, t);
				}else if(field.getType() == Double.class||field.getType() == double.class){
					ReflectUtil.setValueByNameInokeMethod(field,Double.parseDouble(value), t);
				}else if( field.getType() == Float.class||field.getType() == float.class){
					ReflectUtil.setValueByNameInokeMethod(field,Float.parseFloat(value), t);
				}else if(field.getType() == Date.class){
					ReflectUtil.setValueByNameInokeMethod(field , TimeUtil.parseDateString(value, fieldInfo.getFormat()),t);
				}else if(field.getType() == Integer.class||field.getType() == int.class){
					ReflectUtil.setValueByNameInokeMethod(field ,Integer.parseInt(value),t); 
				}else if(field.getType() == Long.class||field.getType() == long.class){
					ReflectUtil.setValueByNameInokeMethod(field ,Long.parseLong(value),t); 
				}else{
					ReflectUtil.setValueByNameInokeMethod(field, value, t);
				}
			}
		}
	}
	/**
	 * 存储批量存储
	 * @param list
	 * @param flushCount put达到阀值后执行持久化操作,该数值为对象成员变量的数量
	 */
	public void save(Collection<T> list,int flushCount){
		if (list == null || list.isEmpty()) {
			return;
		}
		if(flushCount<1){
			flushCount = 500;
		}
		List<Put> puts = new ArrayList<Put>();
		try {
			for (T t : list) {
				puts.addAll(makePuts(t));
				if (puts.size() >= flushCount) {
					HbaseAPI.batchRecord(getTableName(), puts);
					puts.clear();
				}
			}
			if(!puts.isEmpty()){
				HbaseAPI.batchRecord(getTableName(), puts);
			}
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
	}
	/**
	 * 批量存储
	 * @param list
	 */
	public void save(Collection<T> list){
		save(list, 0);
	}
	/**
	 * 组装put
	 * @param t
	 * @return
	 */
	private List<Put> makePuts(T t){
		List<Put> puts = new ArrayList<Put>();
		try{
			for(Entry<String, HFieldInfo> entry : hTableInfo.getBind().entrySet()){
				HFieldInfo fieldInfo = entry.getValue();
				if(fieldInfo!=null){
					String value = "";
					Field field = fieldInfo.getField();
					Object val = ReflectUtil.getValueByNameInokeMethod(field, t);
					if(val != null){
						if(field.getType() == Date.class){
							value = TimeUtil.getDateString((Date)val, fieldInfo.getFormat());
						}else{
							value = val.toString();
						}
					}
					Put p = HbaseAPI.makePut(getTableName(), t.getKey(), hTableInfo.getFamilyName(), entry.getKey(), value);
					if(p != null){
						puts.add(p);
					}
				}
			}
		}catch(Exception e){
			LOG.error(e.getMessage(), e);
		}
		return puts;
	}
	/**
	 * 存储单个实体
	 * @param t
	 */
	public void save(T t){
		if(t == null){
			LOG.warn("entity is null,ignore it");
			return;
		}
		try{
			List<Put> puts = makePuts(t);
			if(!puts.isEmpty()){
				HbaseAPI.batchRecord(getTableName(), puts);
			}
		}catch(Exception e){
			LOG.error(e.getMessage(), e);
		}
	}
	/**
	 * 根据rowkey获取1个实体
	 * @param key
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings("unchecked")
	public T get(byte[] key) throws IOException {
		KeyValue[] keyValue = HbaseAPI.getOneRecord(getTableName(), key);
		T t = null;
		try {
			//实例化
			t = (T) GenericsUtils.getSuperClassGenricType(getClass(),0).newInstance();
			for (KeyValue kv : keyValue) {
				parseKeyValue(kv, t);
			}
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
		return t;
	}
	/**
	 * ScanHbase
	 * @param startRow
	 * @param stopRow
	 * @param filter
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public List<T> scan(String startRow,String stopRow,List<Filter> filter){
		List<T> scanResult = new ArrayList<T>();
		try {
			List<Result> resultList = HbaseAPI.scan(getTableName(), startRow, stopRow, hTableInfo.getFamilyName(), hTableInfo.getBind().keySet(),filter);
			for(Result result : resultList){
				List<KeyValue> keyvalue = result.list();
				T t = (T) GenericsUtils.getSuperClassGenricType(getClass(),0).newInstance();
			 	for(KeyValue kv : keyvalue){
					parseKeyValue(kv, t);
			 	}
			 	scanResult.add(t);
			}
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
		return scanResult;
	}
	/**
	 * ScanHbase
	 * @param startRow
	 * @param stopRow
	 * @param filter
	 * @return
	 */
	public List<T> scan(String startRow,String stopRow){
		return scan(startRow, stopRow, null);
	}
}
